package com.bridgeintelligent.tag.bulkload.service.looklike;

import com.bridgeintelligent.tag.bulkload.api.BulkLoadConfig;
import com.bridgeintelligent.tag.bulkload.api.IndexNameInfo;
import com.bridgeintelligent.tag.bulkload.api.TimeTypeModel;
import com.bridgeintelligent.tag.bulkload.api.fileload.FileLoadModel;
import com.bridgeintelligent.tag.bulkload.api.fileload.FilterGroup;
import com.bridgeintelligent.tag.bulkload.service.core.fileload.ConditionBuilder;
import com.bridgeintelligent.tag.bulkload.service.es.ESUtils;
import com.bridgeintelligent.tag.bulkload.service.es.query.EsQueryBuilder;
import com.bridgeintelligent.tag.constants.bulkload.LookLikeMqMsg;
import com.bridgeintelligent.tag.mapper.DimDataMapper;
import com.bridgeintelligent.tag.mapper.LookLikeMapper;
import com.bridgeintelligent.tag.utils.FieldHandler;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.elasticsearch.common.Strings;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;

@EnableAsync
@Slf4j
@Service
public class AsyncWriteDataService {

    @Autowired
    private ConditionBuilder conditionBuilder;

    @Autowired
    private LookLikeMapper lookLikeMapper;

    @Autowired
    private DimDataMapper dimDataMapper;

    @Autowired
    private HdfsService hdfsService;
    @Autowired
    private ObjectMapper objectMapper;

    @Async
    public void lookalikeDateToHdfs(LookLikeMqMsg lookLikeMqMsg){
        boolean flag = true;
        try {
            //TODO 测试环境不能用,暂时不能测试(已测通)
            TimeTypeModel timeType;
            //如果不属于客群池
            if (!lookLikeMqMsg.getType().equals(ChoiceEnum.CHOICE_THIRD.getCode())){
                //通过客群ID查询indexId和timeType
                timeType = lookLikeMapper.findCustomerTimeType(lookLikeMqMsg.getCustomerId());
            }else{
                timeType = lookLikeMapper.findLookLikeTimeType(lookLikeMqMsg.getId());
            }
            QueryBuilder queryBuilder;
            //如果是动态客群
            if (FieldHandler.FALSE.equals(timeType.getTimeType())) {
                FilterGroup filterGroup = null;
                if (!lookLikeMqMsg.getType().equals(ChoiceEnum.CHOICE_THIRD.getCode())){
                    filterGroup = conditionBuilder.findFilterGroup("-", lookLikeMqMsg.getCustomerId(), new ArrayList<>()).get(0);
                }else{
                    filterGroup = objectMapper.readValue(lookLikeMapper.findFilterGroupById(lookLikeMqMsg.getId()), FilterGroup.class);
                }
                queryBuilder = EsQueryBuilder.buildFilter(filterGroup);
            }else{
                queryBuilder = QueryBuilders.termsQuery(FieldHandler.CUSTOMER_FLAG,timeType.getCustomerId());
            }

            FileLoadModel fileLoad = createFileLoad(queryBuilder);
            IndexNameInfo indexNameInfo = dimDataMapper.findIndexInfoById(timeType.getIndexId());

            String fileName = "";
            if (lookLikeMqMsg.getType().equals("00")){
                fileName = lookLikeMqMsg.getEsIndex()+"_"+"seed.txt";
            }else if (lookLikeMqMsg.getType().equals("01")){
                fileName = lookLikeMqMsg.getEsIndex()+"_"+"black.txt";
            }else if(lookLikeMqMsg.getType().equals("02")){
                fileName = lookLikeMqMsg.getEsIndex()+"_"+"white.txt";
            }else{
                fileName = lookLikeMqMsg.getEsIndex()+"_"+"customer.txt";
            }
            fileName = hdfsService.getPath(fileName,lookLikeMqMsg.getEsIndex());
            FileSystem fileSystem = HdfsUtil.getFileSystem();
            long hdfstartTime = System.currentTimeMillis();    //获取ES查询开始时间
            try (FSDataOutputStream out = fileSystem.create( new Path( fileName ) )){
                ESUtils.scrollSearch(indexNameInfo.getDataDate(), fileLoad, hits -> {
                    StringBuffer sb = new StringBuffer();
                    int i = 0;
                    for (SearchHit hit : hits) {
                        Object s = hit.getSourceAsMap().get(BulkLoadConfig.CUST_NO);
                        if (s == null) {
                            continue;
                        }
                        i++;
                        sb.append(s);
                        sb.append("\n");
                        if(i == 5000){
                            out.writeBytes(sb.toString());
                            sb.setLength(0);
                        }
                    }
                    out.writeBytes(sb.toString());
                });
            }catch (IOException e) {
                log.error("hdfs写文件"+fileName+"失败",e);
                throw e;
            }
            long hdfendTime = System.currentTimeMillis();
            log.info("输出hdf写入运行时间：" + (hdfendTime - hdfstartTime)/1000 + "s");
        } catch (Exception e) {
            log.error("相似客群："+lookLikeMqMsg.getId()+"-文件类型："+lookLikeMqMsg.getType()+"-写入hdfs失败",e);
            updateStatus(lookLikeMqMsg.getType(),AnalysisEnum.ANALYSIS_FOUR.getCode(),ChoiceEnum.CHOICE_THIRD.getCode(),lookLikeMqMsg.getId(),lookLikeMqMsg.getEsIndex());
            flag =false;
        }
        if(flag){
            //如果成功只需改对应的数据文件状态为成功
            updateStatus(lookLikeMqMsg.getType(),"",ChoiceEnum.CHOICE_TWO.getCode(),lookLikeMqMsg.getId(),lookLikeMqMsg.getEsIndex());
        }
    }

    private FileLoadModel createFileLoad(QueryBuilder queryBuilder) {
        FileLoadModel fileLoadModel = new FileLoadModel();
        fileLoadModel.setQueryBuilder(queryBuilder);
        fileLoadModel.setExcludes(Strings.EMPTY_ARRAY);
        fileLoadModel.setIncludes(new String[]{BulkLoadConfig.CUST_NO});
        return fileLoadModel;
    }


    private void updateStatus(String type,String isAnalysis,String status,String id,String esIndex){
        Map<String,String> param = new HashMap<>();
        param.put("id",id);
        param.put("esIndex",esIndex);
        param.put("isAnalysis",isAnalysis);
        switch (type) {
            case "00":
                param.put("seedStatus", status);
                break;
            case "01":
                param.put("blackSeedStatus", status);
                break;
            case "02":
                param.put("whiteSeedStatus", status);
                break;
            default:
                param.put("customerSeedStatus", status);
                break;
        }
        lookLikeMapper.updateDataStatus(param);
    }
}
